package com.study.kafka;

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

/**
 * @author yuantongqin
 * 2021/8/4
 **/
public class ConsumerMain {

    public static Consumer<String,String> createConsumer(){
        Properties properties = new Properties();
        // 设置broker的地址
        properties.put("bootstrap.servers","172.24.36.103:9092");
        // 消费者组id
        properties.put("group.id","demo-consumer-group");
        // 设置消费者分组最初的消费进度 earliest ，lasted,none
        properties.put("auto.offset.reset","earliest");
        // 是否自动提交消费进度，提交offset值
        properties.put("enable.auto.commit",true);
        // 自动提交消费进度频率
        properties.put("auto.commit.interval.ms",1000);
        // consumerConfig 是生产者的配置文件信息
//        ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG
        // 分区者指定策略
//        ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG;
        properties.put("key.deserializer", StringDeserializer.class.getName());
        properties.put("value.deserializer", StringDeserializer.class.getName());

        //创建消费者,消息的key与value与生产者producer保持一致
        return new KafkaConsumer<>(properties);
    }

    public static void main(String[] args) {

        // 创建kafkaConsumer对象
        Consumer<String, String> consumer = createConsumer();
        // 消费者订阅按些topic，可以订阅多个topic但是建议定义一个
        consumer.subscribe(Collections.singleton("TestTopic"));

        // 拉取消息
        while (true){
            ConsumerRecords<String, String> polls = consumer.poll(Duration.ofSeconds(10));
            polls.forEach(new java.util.function.Consumer<ConsumerRecord<String, String>>() {
                @Override
                public void accept(ConsumerRecord<String, String> message) {
                    StringBuilder sb = new StringBuilder();
                    sb.append("消息内容key:").append(message.key());
                    sb.append(" topic: ").append(message.topic());
                    sb.append(" value: ").append(message.value());
                    sb.append(" offset: ").append(message.offset());
                    System.out.println(sb.toString());
                }
            });
            // 设置kafka的手动提交
//            consumer.commitSync();
        }

    }


}
